package com.hehao.reactorPractice;

import com.hehao.entity.BlockingRepository;
import com.hehao.entity.User;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

/**
 * @Author Damon
 * @Date 2024/5/8 17:06
 */
public class BlockingToReactivePractice {
    // TODO Create a Flux for reading all users from the blocking repository deferred until the flux is subscribed, and run it with a bounded elastic scheduler
    //创建一个 Flux，从阻塞存储库中读取所有用户，直到 Flux 被订阅时才执行，并使用有界的弹性调度器运行它。
    Flux<User> blockingRepositoryToFlux(BlockingRepository<User> repository) {
        return Flux.defer(() -> Flux.fromIterable(repository.findAll().toIterable())
                        .filter(User.class::isInstance)
                        .cast(User.class))
                .subscribeOn(Schedulers.boundedElastic());
    }

//========================================================================================

    // TODO Insert users contained in the Flux parameter in the blocking repository using a bounded elastic scheduler and return a Mono<Void> that signal the end of the operation
    //使用有界的弹性调度器将包含在 Flux 参数中的用户插入到阻塞存储库中，并返回一个表示操作结束的 Mono<Void>。
    Mono<Void> fluxToBlockingRepository(Flux<User> flux, BlockingRepository<User> repository) {
        return flux.publishOn(Schedulers.boundedElastic())
//                .doOnNext(repository::save)
                .then();
    }
}
